package com.github.facadefs.oss;

import com.aliyun.oss.ClientException;
import com.aliyun.oss.OSSClient;
import com.aliyun.oss.OSSErrorCode;
import com.aliyun.oss.OSSException;
import com.aliyun.oss.model.BucketStat;
import com.aliyun.oss.model.DeleteObjectsRequest;
import com.aliyun.oss.model.ListObjectsRequest;
import com.aliyun.oss.model.OSSObject;
import com.aliyun.oss.model.OSSObjectSummary;
import com.aliyun.oss.model.ObjectListing;
import com.aliyun.oss.model.ObjectMetadata;
import com.github.facadefs.FileView;
import com.github.facadefs.util.Operation;
import com.github.facadefs.util.Throwables;
import com.google.common.collect.Lists;
import com.github.facadefs.Operations;
import com.github.facadefs.Selector;
import com.github.facadefs.util.Paths;

import java.io.ByteArrayInputStream;
import java.io.DataOutputStream;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.nio.file.FileAlreadyExistsException;
import java.util.Collections;
import java.util.List;

/**
 * 基于阿里云 OSS 的文件系统实现.
 * <p/>
 * 1. 开通 OSS <br/>
 * 2. 创建 bucket <br/>
 * 3. 为 bucket 绑定域名, eg: img.ponly.com --&gt; ponly.oss-cn-hongkong.aliyuncs.com <br/>
 *
 * @author vacoor
 * @see <a href="https://github.com/alibaba-archive/aliyun-oss-hadoop-fs/blob/master/hadoop-tools/hadoop-aliyun-oss/src/main/java/org/apache/hadoop/fs/oss/OSSFileSystem.java">OSSFileSystem</a>
 */
public class AliyunOssFileSystem implements OssFileSystem {
    private static final int DEFAULT_PERMISSIONS = 0600;

    private final String endpoint;
    private final String accessKeyId;
    private final String secretAccessKey;
    private final String bucket;

    private final String chroot;

    public AliyunOssFileSystem(final String endpoint, final String accessKeyId,
                               final String secretAccessKey, final String bucket) {
        this(endpoint, accessKeyId, secretAccessKey, bucket, null);
    }

    public AliyunOssFileSystem(final String endpoint, final String accessKeyId,
                               final String secretAccessKey, final String bucket, final String chroot) {
        this.endpoint = endpoint;
        this.accessKeyId = accessKeyId;
        this.secretAccessKey = secretAccessKey;
        this.bucket = bucket;
        this.chroot = toAbsoluteKey(null, chroot);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public boolean exists(final String path) throws IOException {
        return null != stat(path);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public FileView stat(final String path) throws IOException {
        final String key = toAbsoluteKey(path);
        final String bucket = getBucket(key);
        try {
            final OSSClient oss = connect();
            if (isRootKey(key)) {
                final BucketStat bucketStat = oss.getBucketStat(bucket);
                return toView(bucket, bucketStat);
            }

            final OSSObjectSummary metadata = stat1(oss, bucket, key);
            return null != metadata ? toView(metadata) : null;
        } catch (final Exception ex) {
            return Throwables.rethrowIOException(ex);
        }
    }

    private ObjectMetadata stat0(final OSSClient oss, final String bucket, final String key) throws ClientException, OSSException {
        try {
            return oss.getObjectMetadata(bucket, key);
        } catch (final OSSException oe) {
            if (OSSErrorCode.NO_SUCH_KEY.equals(oe.getErrorCode())) {
                return null;
            }
            throw oe;
        }
    }

    private OSSObjectSummary stat1(final OSSClient oss, final String bucket, final String key) {
        final ObjectMetadata stat = stat0(oss, bucket, key);
        if (null != stat) {
            return toSummary(bucket, key, stat);
        }

        // try as directory
        String directoryKey = key;
        if (isObjectKey(key)) {
            directoryKey = toDirectoryKey(key);
            final ObjectMetadata directoryStat = stat0(oss, bucket, directoryKey);
            if (null != directoryStat) {
                return toSummary(bucket, directoryKey, directoryStat);
            }
        }

        final ListObjectsRequest request = new ListObjectsRequest(bucket);
        request.setPrefix(directoryKey);
        request.setDelimiter("/");
        request.setMaxKeys(1);

        final ObjectListing listing = oss.listObjects(request);
        if (!listing.getCommonPrefixes().isEmpty() || !listing.getObjectSummaries().isEmpty()) {
            final ObjectMetadata metadata = new ObjectMetadata();
            metadata.setContentLength(0);
            return toSummary(bucket, directoryKey, metadata);
        }
        return null;
    }

    private OSSObjectSummary toSummary(final String bucket, final String key, final ObjectMetadata metadata) {
        final OSSObjectSummary summary = new OSSObjectSummary();
        summary.setBucketName(bucket);
        summary.setKey(key);
        summary.setETag(metadata.getETag());
        summary.setSize(metadata.getContentLength());
        summary.setLastModified(metadata.getLastModified());
        summary.setOwner(null);
        summary.setType(metadata.getObjectType());
        summary.setStorageClass(metadata.getObjectStorageClass().toString());
        return summary;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public FileView[] ls(final String path) throws IOException {
        final List<FileView> views = Lists.newLinkedList();
        ls(path, new Selector() {
            @Override
            public int select(final FileView view) {
                views.add(view);
                return CONTINUE;
            }
        });
        return views.toArray(new FileView[0]);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void ls(final String path, final Selector selector) throws IOException {
        final String key = toAbsoluteKey(path);
        final String bucket = getBucket(key);

        try {
            final OSSClient oss = connect();
            if (isObjectKey(key)) {
                final ObjectMetadata stat = stat0(oss, bucket, key);
                if (null != stat) {
                    selector.select(toView(bucket, key, stat));
                    return;
                }
            }

            // root or as directory.
            final String prefix = !isRootKey(key) ? toDirectoryKey(key) : null;
            String marker = null;
            do {
                final ListObjectsRequest request = new ListObjectsRequest();
                request.setBucketName(bucket);
                request.setPrefix(prefix);
                request.setMarker(marker);
                request.setDelimiter("/");
                // request.setMaxKeys(1000);

                final ObjectListing listing = oss.listObjects(request);
                final List<OSSObjectSummary> objects = listing.getObjectSummaries();
                for (final OSSObjectSummary object : objects) {
                    if (Selector.CONTINUE != selector.select(toView(object))) {
                        break;
                    }
                }

                final List<String> directoryKeys = listing.getCommonPrefixes();
                for (final String directoryKey : directoryKeys) {
                    if (Selector.CONTINUE != selector.select(toView(bucket, directoryKey))) {
                        break;
                    }
                }
                marker = listing.getNextMarker();
            } while (null != marker && !marker.isEmpty());
        } catch (final Exception ex) {
            Throwables.rethrowIOException(ex);
        }
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public boolean mkdirs(String path) throws IOException {
        final String key = toAbsoluteKey(path);
        final String bucket = getBucket(key);
        try {
            final String directoryKey = toDirectoryKey(key);
            final OSSClient oss = connect();
            oss.putObject(bucket, directoryKey, new ByteArrayInputStream(new byte[0]));
            return true;
        } catch (final Exception ex) {
            return Throwables.rethrowIOException(ex);
        }
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void rm(final String path, boolean recursive) throws IOException {
        final String key = toAbsoluteKey(path);
        final String bucket = getBucket(key);

        try {
            final OSSClient oss = connect();
            if (isObjectKey(key) && oss.doesObjectExist(bucket, key)) {
                final List<String> deletedKeys = deleteObjects0(oss, bucket, Collections.singletonList(key));
                if (0 < deletedKeys.size()) {
                    return;
                }
            }

            // 删除失败或不是文件
            final String prefix = !isRootKey(key) ? toDirectoryKey(key) : null;
            final ListObjectsRequest request = new ListObjectsRequest(bucket, prefix, null, null, 1000);

            String marker = null;
            do {
                final ObjectListing listing = oss.listObjects(request);
                final List<OSSObjectSummary> objects = listing.getObjectSummaries();
                // final List<String> directories = listing.getCommonPrefixes();
                if (!objects.isEmpty()) {
                    if (!recursive) {
                        throw new IOException("You cannot delete non-empty directory, use recurse=true to overide");
                    }

                    final List<String> keysToDelete = Lists.newArrayListWithExpectedSize(objects.size());
                    for (final OSSObjectSummary object : objects) {
                        keysToDelete.add(object.getKey());
                    }
                    final List<String> deletedKeys = deleteObjects0(oss, bucket, keysToDelete);
                    keysToDelete.removeAll(deletedKeys);

                    if (!keysToDelete.isEmpty()) {
                        throw new IOException("cannot delete objects: " + keysToDelete);
                    }
                }

                marker = listing.getNextMarker();
            } while (null != marker && !marker.isEmpty());

            if (null != prefix) {
                List<String> deletedKeys = deleteObjects0(oss, bucket, Collections.singletonList(prefix));
                if (deletedKeys.isEmpty()) {
                    throw new IOException("cannot delete directory: " + prefix);
                }
            }
        } catch (final Exception ex) {
            Throwables.rethrowIOException(ex);
        }
    }

    private List<String> deleteObjects0(final OSSClient oss, final String bucket, final List<String> keys) {
        final DeleteObjectsRequest delRequest = new DeleteObjectsRequest(bucket);
        delRequest.setKeys(keys);
        return oss.deleteObjects(delRequest).getDeletedObjects();
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void rename(String srcPath, String newPath) throws IOException {
        final String srcKey = toAbsoluteKey(srcPath);
        final String dstKey = toAbsoluteKey(newPath);

        final String srcBucket = getBucket(srcKey);
        final String dstBucket = getBucket(dstKey);

        final OSSClient oss = connect();
        if (!oss.doesObjectExist(srcBucket, srcKey)) {
            throw new FileNotFoundException(srcPath);
        }
        if (!oss.doesObjectExist(dstBucket, dstKey)) {
            throw new FileAlreadyExistsException(dstKey);
        }

        // FIXME 可能是目录重命名
        oss.copyObject(srcBucket, srcKey, dstBucket, dstKey);
        oss.deleteObject(srcBucket, srcKey);
    }


    /**
     * {@inheritDoc}
     */
    @Override
    public InputStream open(final String path) throws IOException {
        final String key = toAbsoluteKey(path);
        final String bucket = getBucket(key);
        try {
            final OSSClient oss = connect();
            final OSSObjectSummary stat = stat1(oss, bucket, key);
            if (null == stat) {
                return null;
            } else if (isDirectoryKey(stat.getKey())) {
                throw new IOException("Can't open " + path + " because it is a directory");
            }

            final OSSObject object = getObject0(oss, bucket, key);
            return null != object ? object.getObjectContent() : null;
        } catch (final Exception ex) {
            return Throwables.rethrowIOException(ex);
        }
    }

    private OSSObject getObject0(final OSSClient oss, final String bucket, final String key)
            throws ClientException, OSSException {
        try {
            return oss.getObject(bucket, key);
        } catch (final OSSException ex) {
            if (OSSErrorCode.NO_SUCH_KEY.equals(ex.getErrorCode())) {
                return null;
            }
            throw ex;
        }
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public OutputStream create(String path, boolean override) throws IOException {
        return create(path, null, override);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public OutputStream create(final String path, final String contentType, final boolean override) throws IOException {
        final String key = toAbsoluteKey(path);
        final String bucket = getBucket(key);

        try {
            final OSSClient oss = connect();
            final OSSObjectSummary stat = stat1(oss, bucket, key);
            if (null != stat) {
                if (!override) {
                    throw new IOException("File already exists: " + path);
                } else if (isDirectoryKey(stat.getKey())) {
                    throw new IOException("Can't override " + path + " because it is a directory");
                }
            }

            final File tempFile = File.createTempFile("oss.upload.", ".tmp");
            return new DataOutputStream(new FileOutputStream(tempFile)) {
                @Override
                public void close() throws IOException {
                    super.close();
                    try {
                        final ObjectMetadata metadata = new ObjectMetadata();
                        if (null != contentType) {
                            metadata.setContentType(contentType);
                        }
                        oss.putObject(bucket, key, tempFile, metadata);
                    } finally {
                        if (!tempFile.delete()) {
                            tempFile.deleteOnExit();
                        }
                    }
                }
            };
        } catch (final Exception ex) {
            return Throwables.rethrowIOException(ex);
        }
    }

    @Override
    public <R> R execute(final Operation<Operations, R> operation) throws IOException {
        return operation.execute(this);
    }

    protected String getBucket(String absPath) {
        return bucket;
    }

    protected OSSClient connect() {
        return new OSSClient(endpoint, accessKeyId, secretAccessKey);
    }

    private boolean isRootKey(final String key) {
        return null == key || key.isEmpty();
    }

    private boolean isObjectKey(final String key) {
        return null != key && !key.isEmpty() && !key.endsWith(Paths.PATH_SEPARATOR);
    }

    private boolean isDirectoryKey(final String key) {
        return null != key && !key.isEmpty() && key.endsWith(Paths.PATH_SEPARATOR);
    }

    private String toDirectoryKey(final String key) {
        return null != key && !key.endsWith(Paths.PATH_SEPARATOR) ? key + Paths.PATH_SEPARATOR : key;
    }

    private boolean isDirectory(final String key, final long size) {
        return isDirectoryKey(key) && size == 0L;
    }

    protected String toAbsoluteKey(String path) {
        return toAbsoluteKey(chroot, path);
    }

    private String toAbsoluteKey(final String chroot, final String path) {
        if (null == path) {
            throw new IllegalArgumentException("path muse be not empty");
        }
        if (null != chroot && !chroot.isEmpty()) {
            return Paths.toAbsolutePath(chroot, path);
        } else if (path.startsWith(Paths.ROOT)) {
            return Paths.toRelativePath(path, Paths.ROOT);
        }
        return path;
    }

    private String toRelativeKey(final String absKey) {
        if (null != chroot && !chroot.isEmpty()) {
            return Paths.toRelativePath(chroot, absKey);
        }
        return absKey;
    }

    private FileView toView(final String bucket, final BucketStat stat) {
        return new FileView(
                Paths.ROOT, stat.getStorageSize(), true,
                stat.getLastModifiedTime(), -1, DEFAULT_PERMISSIONS,
                null, bucket, null
        );
    }

    private FileView toView(final String bucket, final String key) {
        return new FileView(toRelativeKey(key), -1, true, -1, -1, DEFAULT_PERMISSIONS, null, null, null);
    }

    private FileView toView(final String bucket, final String key, final ObjectMetadata metadata) {
        final boolean isObjectKey = isObjectKey(key);
        final long length = metadata.getContentLength();
        final long mtime = metadata.getLastModified().getTime();
        return new FileView(
                toRelativeKey(key), length, !isObjectKey, mtime, -1,
                DEFAULT_PERMISSIONS, null, bucket, null
        );
    }


    private FileView toView(final OSSObjectSummary summary) {
        if (null != summary) {
            final String key = summary.getKey();
            final long size = summary.getSize();
            final long mtime = summary.getLastModified().getTime();
            final String owner = summary.getOwner().getDisplayName();
            final String group = summary.getBucketName();

            return new FileView(toRelativeKey(key), size, false, mtime, mtime, DEFAULT_PERMISSIONS, owner, group, null);
        }
        return null;
    }

}
